package org.apache.flink.connector.httptm;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.RuntimeContextInitializationContextAdapters;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.httptm.factories.DefaultRequestFactory;
import org.apache.flink.connector.httptm.factories.RequestFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.data.RowData;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.ServiceLoader;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <ol>
 * date:2022/4/6 editor:yanghongjian
 * <li>创建文档</li>
 * </ol>
 * <ol>
 *
 * @author <a href="mailto:12719889@qq.com">YangHongJian</a>
 * @version 3.0
 * @since 1.8
 */
public class HttpTmSourceFunction extends RichSourceFunction<RowData> implements ResultTypeQueryable<RowData> {
    private static final Logger logger = LoggerFactory.getLogger(HttpTmSourceFunction.class);
    private volatile boolean isRunning;
    private final ReadableConfig options;
    private final DeserializationSchema<RowData> deserializer;
    private final boolean isBounded;

    private final ObjectMapper objectMapper = new ObjectMapper();

    public HttpTmSourceFunction(
            ReadableConfig options,
            DeserializationSchema<RowData> deserializer,
            boolean isBounded) {
        this.options = options;
        this.deserializer = deserializer;
        this.isBounded = isBounded;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        deserializer.open(
                RuntimeContextInitializationContextAdapters.deserializationAdapter(
                        getRuntimeContext()));
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return deserializer.getProducedType();
    }

    @Override
    public void run(SourceContext<RowData> sourceContext) throws Exception {
        //该方法只执行一次,而构造器多次执行
        String headers = options.get(HttpTmTableSourceFactory.HEADERS),
                queryParams = options.get(HttpTmTableSourceFactory.QUERYPARAMS),
                body = options.get(HttpTmTableSourceFactory.BODY),
                setting = options.get(HttpTmTableSourceFactory.SETTING),
                url = options.get(HttpTmTableSourceFactory.URL),
                method = options.get(HttpTmTableSourceFactory.METHOD);
        Objects.requireNonNull(url, "请求地址不允许为空!");
        Objects.requireNonNull(method, "请求类型不允许为空!");


        Map<String, String> mapHeaders = objectMapper.readValue(headers, Map.class);
        Map<String, String> mapQueryParams = objectMapper.readValue(queryParams, Map.class);
        Map<String, Object> mapSetting = objectMapper.readValue(setting, Map.class);

        int interval = Integer.parseInt(mapSetting.getOrDefault("interval", 10).toString()) * 1000,
                timeout = Integer.parseInt(mapSetting.getOrDefault("timeout", 0).toString()) * 1000;


        AtomicInteger size = new AtomicInteger();
        Collector<RowData> out = new Collector<RowData>() {
            @Override
            public void collect(RowData record) {
                size.incrementAndGet();
                sourceContext.collect(record);
            }

            @Override
            public void close() {
            }
        };

        String authorType = (String) mapSetting.getOrDefault("factory", "");
        RequestInfo requestInfo = getRequestStrategy(authorType).createRequestInfo(url, mapHeaders, mapQueryParams, body, mapSetting);

        this.isRunning = true;
        while (isRunning) {
            try {
                byte[] byteRet = HttpTools.httpRequest(requestInfo.getURI(), method, requestInfo.getHeaders(), requestInfo.getBodyBytes(StandardCharsets.UTF_8), timeout, timeout);
                if (Objects.nonNull(byteRet) && byteRet.length > 0) {
                    deserializer.deserialize(byteRet, out);
                }
                //批模式时执行一次
                if (interval <= 0 || isBounded) {
                    isRunning = false;
                } else if (isRunning) {
                    Thread.sleep(interval);
                }
            } catch (Exception e) {
                logger.error("http接口采集异常", e);
                throw new RuntimeException(e.getMessage());
            }
        }
    }

    /**
     * 获取 自定义策略
     * @param authorType
     * @return
     */
    private RequestFactory getRequestStrategy(String authorType) {
        if(authorType==null || authorType.isEmpty()) {
            return new DefaultRequestFactory();
        }
        ServiceLoader<RequestFactory> factories = ServiceLoader.load(RequestFactory.class);
        for (RequestFactory factory : factories) {
            if (factory.factoryIdentifier().equals(authorType)) {
                return factory;
            }
        }
        throw new RuntimeException("http request factory[" + authorType + "] not support");
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
